package com.example.nettyrpcdemo.core;

import com.google.protobuf.GeneratedMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.InetSocketAddress;
import java.nio.channels.spi.SelectorProvider;
import lombok.extern.slf4j.Slf4j;
import tools.MessageManager;

@Slf4j
public class NettyRpcClient {

    // 共用
    private static final NioEventLoopGroup workerGroup = new NioEventLoopGroup(8, new DefaultThreadFactory("client"), SelectorProvider.provider());

    private Channel channel;
    protected Bootstrap bootstrap;

    public void connect(String ip, int port) throws Exception {
        this.bootstrap = new Bootstrap();
        bootstrap.group(workerGroup).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).channelFactory(NioSocketChannel::new);
        bootstrap.handler(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new Decoder());
                pipeline.addLast(new Encoder());
                pipeline.addLast(new NettyClientHandler());
            }
        });

        ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(ip, port));
        this.channel = channelFuture.sync().channel();
        log.info("NettyRpcClient connect to ip={}, port={} success!", ip, port);
    }

    /**
     * 向服务提供者发送消息
     *
     * @param msg
     * @param seqId
     */
    public void send(GeneratedMessage msg, long seqId) {
        int msgId = MessageManager.getMsgIdByMsgClazz(msg.getClass());

        byte[] bytes = msg.toByteArray();

        // 4head 4msgid + 8seqId + body
        int len = 4 + 4 + 8 + bytes.length;
        ByteBuf byteBuf = channel.alloc().buffer(len);
        byteBuf.writeInt(len);
        byteBuf.writeInt(msgId);
        byteBuf.writeLong(seqId);
        byteBuf.writeBytes(bytes);

        // 编码
        this.channel.writeAndFlush(byteBuf);
    }
}
